// Copyright (C) 2012 Yury Bychenkov (http://bychenkov.com).
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
//  - Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//  - Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
//  - Neither the name of Yury Bychenkov nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

template <typename Provider>
QueryExecutor<Provider>::QueryExecutor(const Query& query_, const Provider& provider_):
	query(query_),
	provider(provider_),
	code(0),
	columnMapping(0),
	columnCount(0),
	rangeFilter(0),
	rangeFilterCount(0)
{
}

template <typename Provider>
QueryExecutor<Provider>::~QueryExecutor() {
	if (code != 0) {
		delete [] code;
	}
	if (pivots != 0) {
		delete [] pivots;
	}                   	
	if (columnMapping != 0) {
		delete [] columnMapping;
	}                                                                         
	if (rangeFilter != 0) {
		delete [] rangeFilter;
	}
}


template <typename Provider>
void QueryExecutor<Provider>::explain(ostream& os) {
	for (uint i = 0; i < rangeFilterCount; i++) {
		if (i > 0) {
			os << ", ";
		}

		uint columnIndex = columnMapping[i];
		uint rangeMin = rangeFilter[2 * columnIndex];
		uint rangeMax = rangeFilter[2 * columnIndex + 1];
		os << query.schema.getColumn(columnIndex).name <<" in [" << rangeMin << ", ";
		if (rangeMax == 0xFFFFFFFFU) {
			os << "...)";
		} else {
			os << rangeMax << "]";
		}
	}

	// TODO
}

template <typename Provider>
QueryExecutor<Provider>* QueryExecutor<Provider>::link(const Query& query, const Provider& provider) {
	QueryExecutor<Provider>* queryExecutor = new QueryExecutor<Provider>(query, provider);

	if (queryExecutor->link()) {
		queryExecutor->buildRangeFilter();
		return queryExecutor;
	}

	delete queryExecutor;

	return 0;
}

template <typename Provider>
template <typename Consumer>
uint QueryExecutor<Provider>::execute(Consumer& consumer, uint8* memory, uint memorySize) const {
	if (code[0] == BC_RET_FALSE) {
		return 0;
	}

	typename Provider::Cursor cursor(provider, columnMapping, columnCount, rangeFilter, rangeFilterCount);

	TupleBuilder tupleBuilder;

	const uint prefixCount = orderingPivotCount;
	uint topCount = query.topCount;

	uint count = 0;
	bool skipMatch = code[0] == BC_RET_TRUE; 

	if (prefixCount == 0) {
		while (cursor.next()) {
			if (skipMatch || matches(cursor)) {
				count++;
	    
				tupleBuilder.reset();
				buildTuple(cursor, tupleBuilder, 0, pivotCount);
				if (!consumer.put(tupleBuilder.build()) || count == topCount) {
					break;
				}
			}
		}
		return count;
	}

	Heap<TupleTraits> tupleHeap(memory, memorySize, topCount);

	while (cursor.next()) {
		if (skipMatch || matches(cursor)) {
			count++;

			tupleBuilder.reset();
			buildTuple(cursor, tupleBuilder, 0, prefixCount);
			if (!tupleHeap.putTest(tupleBuilder.build())) {
				continue;
			}
			buildTuple(cursor, tupleBuilder, prefixCount, pivotCount);
			if (!tupleHeap.put(tupleBuilder.build())) {
				logger.error("Ordering heap overflow");
				break;
			}
		}
	}

	for (uint i = 0, k = tupleHeap.getCount(); i < k; i++) {
		if (!consumer.put(tupleHeap.get(i))) {
			break;
		}
	}

	return count;
}

template <typename Provider>
void QueryExecutor<Provider>::buildTuple(typename Provider::Cursor& cursor, TupleBuilder& tupleBuilder, uint fromIndex, uint toIndex) const {
	for (uint pivotIndex = fromIndex; pivotIndex < toIndex; pivotIndex++) {
		const Pivot& pivot = pivots[pivotIndex];
		if (pivot.useIndex) {
			tupleBuilder.addInt64(cursor.getIndex((int64) pivot.columnIndex), pivot.asc);
		} else {
			switch (pivot.type) {
				case Schema::STRING: {
					uint size;
					const uint8* value = StringTraits::decode(cursor.getString(pivot.columnIndex), size);
					tupleBuilder.addBytes(value, size, pivot.asc);
					break;
				}

				case Schema::INTEGER: {
					int64 value = IntegerTraits::decode(cursor.getInteger(pivot.columnIndex));
					tupleBuilder.addInt64(value, pivot.asc);
					break;
				}

				case Schema::DECIMAL: {
					uint size;
					const uint8* value = DecimalTraits::decode(cursor.getDecimal(pivot.columnIndex), size);
					tupleBuilder.addBytes(value, size, pivot.asc);
					break;
				}
			}
		}
	}
}

template <typename Provider>
bool QueryExecutor<Provider>::link() {
	const ResourceBundle& bundle = query.data;
	const Schema& querySchema = query.schema;
	const Schema& providerSchema = provider.getSchema();
	
	codeCount = query.codeCount;
	code = new uint16[codeCount];
	memcpy(code, query.code, codeCount * sizeof(uint16));
	uint16* p = code;

	orderingPivotCount = query.ordering.size();
	pivotCount = query.ordering.size() + query.projections.size();
	pivots = pivotCount == 0 ? 0 : new Pivot[pivotCount];

	map<uint, uint> columnMap;

	while (true) {
		switch (p[0]) {
			case BC_RET_FALSE:
				// BC_RET_TRUE always goes first if isn't eliminated
				code[0] = BC_RET_FALSE;
				codeCount = 1;
				return true;

			case BC_RET_TRUE:
				pivotCount = 0;
				for (uint i = 0; i < query.ordering.size(); i++) {
					uint columnIndex = query.ordering[i].columnIndex;
					if (columnMap.find(columnIndex) == columnMap.end()) {
						uint mapIndex = columnMap.size();
						columnMap[columnIndex] = mapIndex;
					}
					Pivot& p = pivots[pivotCount++];
					p.columnIndex = columnMap[columnIndex];
					p.type = querySchema.getColumn(columnIndex).type;
					p.useIndex = false;
					p.asc = query.ordering[i].asc;
				}

				for (uint i = 0; i < query.projections.size(); i++) {
					uint columnIndex = query.projections[i].columnIndex;
					if (columnMap.find(columnIndex) == columnMap.end()) {
						uint mapIndex = columnMap.size();
						columnMap[columnIndex] = mapIndex;
					} else {
						int j = int(pivotCount) - 1;
						while (j >= 0 && pivots[j].columnIndex != columnMap[columnIndex]) {
							j--;
						}
						if (j >= 0) {
							continue;
						}
					}

					Pivot& p = pivots[pivotCount++];
					p.columnIndex = columnMap[columnIndex];
					p.type = querySchema.getColumn(columnIndex).type;
					p.useIndex = false;
					p.asc = true;
				}

				if (columnMap.size() == 0) {
					columnMap[0] = 0;
				}

				columnCount = columnMap.size();
				columnMapping = new uint [columnCount];

				for (map<uint, uint>::iterator it = columnMap.begin(); it != columnMap.end(); it++) {
					columnMapping[it->second] = it->first;	
				} 

				return true;

			case BC_EQ_I:
			case BC_LE_I:
			case BC_LT_I:
			case BC_GE_I:
			case BC_GT_I:
			case BC_NE_I: {
				const uint columnIndex = p[1];
				if (columnIndex >= querySchema.getColumnCount() || columnIndex >= providerSchema.getColumnCount()) {
					return false;
				}
				const uint columnType = querySchema.getColumn(columnIndex).type;
				if (providerSchema.getColumn(columnIndex).type != columnType) {
					return false;
				}
				const uint8* value = bundle.get(p[2]);
				if (value == 0) {
					return false;
				}
				if (p[6] != BC_IF) {
					return false;
				}

				
				if (provider.isIndexAvailable(columnIndex)) {
					int indexValue;
					switch (columnType) {
						case Schema::STRING:
							indexValue = provider.find<StringTraits>(columnIndex, value);
							break;
						case Schema::INTEGER:
							indexValue = provider.find<IntegerTraits>(columnIndex, value);
							break;
						case Schema::DECIMAL:
							indexValue = provider.find<DecimalTraits>(columnIndex, value);
							break;
						default: 
							return false;
					}
					if (eliminateInvariant(p, indexValue, (int) provider.getCardinality(columnIndex))) {
						break;
					}
					if (indexValue < 0) {
						// -index - 1.5
						if (p[0] == BC_LE_I || p[0] == BC_GT_I) {
							indexValue = -indexValue - 2;
						} else if (p[0] == BC_LT_I || p[0] == BC_GE_I) {
							indexValue = -indexValue - 1;
						} else {
							return false; // never happens
						}
					}
					*(uint32*)(p + 2) = (uint32) indexValue;
				} else {
					switch (columnType) {
						case Schema::STRING:
							p[0] += BC_EQ_I_S - BC_EQ_I;
							*(const uint8**) (p + 2) = value;
							break;
						case Schema::INTEGER:
							p[0] += BC_EQ_I_L - BC_EQ_I;
							*(int64*) (p + 2) = IntegerTraits::decode(value);
							break;
						case Schema::DECIMAL:
							p[0] += BC_EQ_I_D - BC_EQ_I;
							*(const uint8**) (p + 2) = value;
							break;
						default:
							return false;
					}
				}

				if (columnMap.find(columnIndex) == columnMap.end()) {
					uint mapIndex = columnMap.size();
					columnMap[columnIndex] = mapIndex;
				}
				p[1] = (uint16) columnMap[columnIndex];

				p += 8; // BC_xx_xx + BC_IF : 6 + 2

				break;
			}
	        
			default:
				return false;
		}
	}
}

template <typename Provider>
bool QueryExecutor<Provider>::eliminateInvariant(const uint16* p, int indexValue, int indexCardinality) {
	bool invariant;

	switch (p[0]) {
		case BC_EQ_I:
			if (indexValue < 0) {
				invariant = false;
				break;
			}
			return false;

		case BC_LE_I:
			if (indexValue == indexCardinality - 1 || -indexValue - 1 == indexCardinality) {
			    invariant = true;
			    break;
		    }
		    return false;

		case BC_LT_I:
			if (indexValue == 0) {
			    invariant = false;
			    break;
		    } else if (-indexValue - 1 == indexCardinality) {
		    	invariant = true;
		    	break;
		    }
		    return false;

		case BC_GE_I:
			if (indexValue == 0) {
			    invariant = true;
			    break;
		    } else if (-indexValue - 1 == indexCardinality) {
		    	invariant = false;
		    	break;
		    }
		    return false;

		case BC_GT_I:
			if (indexValue == indexCardinality - 1 || -indexValue - 1 == indexCardinality) {
			    invariant = false;
			    break;
		    }
		    return false;

		case BC_NE_I: 
			if (indexValue < 0) {
				invariant = true;
				break;
			}
			return false;

		default:
			return false;
	}
	
	if (p[6] != BC_IF){
		return false;
	}

	eliminateCodeBlock(p - code, 6 + (invariant ? p[7] : 2));
	
	return true;
}

template <typename Provider>
void QueryExecutor<Provider>::eliminateCodeBlock(uint start, uint size) {
	memmove(code + start, code + start + size, (codeCount - start - size) * sizeof(uint16));
	codeCount -= size;

	for (uint p = 0; p < start; ) {
	 	uint16 opcode = code[p];
	 	if (opcode == BC_IF){
	 	    if (code[p + 1] + p > start) {
	 	    	code[p + 1] -= size;
	 	    }
	 	    p += 2;
	 	} else if (opcode == BC_RET_FALSE || opcode == BC_RET_TRUE) {
	 		p++; // never happens
	 	} else { // BC_cmp_x
	 		p += 6;
	 	}
	}	
}

template <typename Provider>
void QueryExecutor<Provider>::buildRangeFilter() {
	rangeFilter = new uint32[columnCount * 2];

	const uint16* p = code;
	while (true) {
		const uint opcode = p[0];
		if (!(BC_EQ_I <= opcode && opcode <= BC_NE_I && p[6] == BC_IF && p[6 + p[7]] == BC_RET_FALSE)) {
			break;
		}
		const uint columnIndex = p[1];
		if (columnIndex > rangeFilterCount) {
			break; // never happens
		}

		uint32& rangeMin = rangeFilter[columnIndex * 2];
		uint32& rangeMax = rangeFilter[columnIndex * 2 + 1];
		if (columnIndex == rangeFilterCount) {
			rangeFilterCount++;
			rangeMin = 0;
			rangeMax = 0xFFFFFFFFU;
		}

		uint32 indexValue = *(uint32*)(p + 2);
		bool eliminate = true;

		switch (opcode) {
			case BC_EQ_I:
				if (rangeMin == indexValue) {
					rangeMin = indexValue + 1;
				} else if (rangeMax == indexValue) {
					rangeMax = indexValue - 1;
				} else {
					eliminate = false;
				}
				break;
	    
			case BC_LE_I:
				rangeMin = max(rangeMin, indexValue + 1);
			    break;
	    
			case BC_LT_I:
				rangeMin = max(rangeMin, indexValue);
			    break;
	    
			case BC_GE_I:
				if (indexValue == 0) {
					rangeMin = 1;
					rangeMax = 0;
				} else {
					rangeMax = min(rangeMax, indexValue - 1);
				}
			    break;
	    
			case BC_GT_I:
				rangeMax = min(rangeMax, indexValue);
			    break;
	    
			case BC_NE_I:
				rangeMin = max(rangeMin, indexValue);
				rangeMax = min(rangeMax, indexValue);
				break; 
		}

		if (rangeMin > rangeMax) {
			code[0] = BC_RET_FALSE;
			codeCount = 1;
			rangeFilterCount = 0;
			return;
		}

		if (eliminate) {
			eliminateCodeBlock(p - code, 8);
		} else {
			p += 8;
		}
	}
}

template <typename Provider>
bool QueryExecutor<Provider>::matches(typename Provider::Cursor& cursor) const {

	const uint16* p = code;

	bool condition = false;

	while (true) {
		switch (p[0]) {
			case BC_RET_FALSE: 
				return false;

			case BC_RET_TRUE: 
				return true;

			case BC_IF: 
				p += condition? p[1] : 2;
				break;

			case BC_EQ_I:
				condition = cursor.getIndex(p[1]) == *(const uint32*)(p + 2);
				p += 6;
				break;
			case BC_LE_I:
				condition = cursor.getIndex(p[1]) <= *(const uint32*)(p + 2);
				p += 6;
				break;
			case BC_LT_I:
				condition = cursor.getIndex(p[1]) < *(const uint32*)(p + 2);
				p += 6;
				break;
			case BC_GE_I:
				condition = cursor.getIndex(p[1]) >= *(const uint32*)(p + 2);
				p += 6;
				break;
			case BC_GT_I:
				condition = cursor.getIndex(p[1]) > *(const uint32*)(p + 2);
				p += 6;
				break;
			case BC_NE_I:
				condition = cursor.getIndex(p[1]) != *(const uint32*)(p + 2);
				p += 6;
				break;


			case BC_EQ_I_S:
				condition = StringTraits::equal(cursor.getString(p[1]), *(const uint8**)(p + 2));
				p += 6;
				break;
			case BC_LE_I_S:
				condition = StringTraits::compare(cursor.getString(p[1]), *(const uint8**)(p + 2)) <= 0;
				p += 6;
				break;
			case BC_LT_I_S:
				condition = StringTraits::compare(cursor.getString(p[1]), *(const uint8**)(p + 2)) < 0;
				p += 6;
				break;
			case BC_GE_I_S:
				condition = StringTraits::compare(cursor.getString(p[1]), *(const uint8**)(p + 2)) >= 0;
				p += 6;
				break;
			case BC_GT_I_S:
				condition = StringTraits::compare(cursor.getString(p[1]), *(const uint8**)(p + 2)) > 0;
				p += 6;
				break;
			case BC_NE_I_S:
				condition = !StringTraits::equal(cursor.getString(p[1]), *(const uint8**)(p + 2));
				p += 6;
				break;

			case BC_EQ_I_L:
				condition = IntegerTraits::decode(cursor.getInteger(p[1])) == *(const int64*)(p + 2);
				p += 6;
				break;
			case BC_LE_I_L:
				condition = IntegerTraits::decode(cursor.getInteger(p[1])) <= *(const int64*)(p + 2);
				p += 6;
				break;
			case BC_LT_I_L:
				condition = IntegerTraits::decode(cursor.getInteger(p[1])) < *(const int64*)(p + 2);
				p += 6;
				break;
			case BC_GE_I_L:
				condition = IntegerTraits::decode(cursor.getInteger(p[1])) >= *(const int64*)(p + 2);
				p += 6;
				break;
			case BC_GT_I_L:
				condition = IntegerTraits::decode(cursor.getInteger(p[1])) > *(const int64*)(p + 2);
				p += 6;
				break;
			case BC_NE_I_L:
				condition = IntegerTraits::decode(cursor.getInteger(p[1])) != *(const int64*)(p + 2);
				p += 6;
				break;

			case BC_EQ_I_D:
				condition = DecimalTraits::equal(cursor.getDecimal(p[1]), *(const uint8**)(p + 2));
				p += 6;
				break;
			case BC_LE_I_D:
				condition = DecimalTraits::compare(cursor.getDecimal(p[1]), *(const uint8**)(p + 2)) <= 0;
				p += 6;
				break;
			case BC_LT_I_D:
				condition = DecimalTraits::compare(cursor.getDecimal(p[1]), *(const uint8**)(p + 2)) < 0;
				p += 6;
				break;
			case BC_GE_I_D:
				condition = DecimalTraits::compare(cursor.getDecimal(p[1]), *(const uint8**)(p + 2)) >= 0;
				p += 6;
				break;
			case BC_GT_I_D:
				condition = DecimalTraits::compare(cursor.getDecimal(p[1]), *(const uint8**)(p + 2)) > 0;
				p += 6;
				break;
			case BC_NE_I_D:
				condition = !DecimalTraits::equal(cursor.getDecimal(p[1]), *(const uint8**)(p + 2));
				p += 6;
				break;

			default:
				return false;

		}
	}
}
